package com.db.concurrency.producerconsumer.threads;

import java.io.File;
import java.io.FileFilter;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * @author Alexander Kotov (kotov.alex.22@gmail.com)
 */
public class ProducerConsumer {
    private CountDownLatch crawlDoneSignal;
    private static final int N_CONSUMERS = Runtime.getRuntime().availableProcessors();
    private final List<BlockingDeque<File>> tasksList = new ArrayList<BlockingDeque<File>>();
    private final List<Thread> indexers = new ArrayList<Thread>();

    public void startIndexing(File[] roots) {
        crawlDoneSignal = new CountDownLatch(roots.length);

        FileFilter filter = new FileFilter() {
            public boolean accept(File file) {
                return true;
            }
        };

        for (int i = 0; i < N_CONSUMERS; ++i) {
            tasksList.add(new LinkedBlockingDeque<File>());
        }

        int curId = 0;
        for (File root : roots) {
            new Thread(new FileCrawler(tasksList, curId, filter, root, crawlDoneSignal)).start();
            curId = ++curId % N_CONSUMERS;
        }

        for (int i = 0; i < N_CONSUMERS; ++i) {
            Thread thread = new Thread(new FileIndexer(i, tasksList));
            thread.start();
            indexers.add(thread);
        }
    }

    public void awaitCrawling() throws InterruptedException {
        System.out.println("waiting...");
        crawlDoneSignal.await();
        System.out.println("crawling finished.");
    }

    public void stopIndexing() {
        for (Thread thread : indexers) {
            thread.interrupt();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        File file = new File(args[0]);
        ProducerConsumer producerConsumer = new ProducerConsumer();

        producerConsumer.startIndexing(file.listFiles());

        producerConsumer.awaitCrawling();

        producerConsumer.stopIndexing();
    }
}
